package cn.lsh.hbase.coprocessor;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.coprocessor.ColumnInterpreter;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * AggregationClient提供了一系列的hbase表的聚合函数，用于hbase的聚合操作，需要protobuf低于3.0版本的依赖
 * 启动协处理器 Aggregation有以下两种方法：
 * 		1、修改hbase-site.xml配置文件，启动全局aggregation，能操纵所有的表上的数据
 * 			<property>
 *    			<name>hbase.coprocessor.user.region.classes</name>
 *    			<value>org.apache.hadoop.hbase.coprocessor.AggregateImplementation</value>
 *  		</property>
 *  	2、启用表aggregation，只对特定的表生效。
 *  			(1)disable指定表。hbase> disable 'total'
 * 				(2)添加aggregation hbase> alter 'total', METHOD => 'table_att','coprocessor'=>'|org.apache.hadoop.hbase.coprocessor.AggregateImplementation||'
 * 				(3)重启指定表 hbase> enable 'total'
 */
public class AggregationTest {

	public static void main(String[] args) throws Throwable {
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.set("hbase.zookeeper.quorum", "node01,node02,node03");
		//提高RPC通信时长
		conf.setLong("hbase.rpc.timeout", 600000);
		//设置Scan缓存
		conf.setLong("hbase.client.scanner.caching", 1000);
		//启动全局aggregation，能过操纵所有的表上的数据
		// conf.set("hbase.coprocessor.user.region.classes", "org.apache.hadoop.hbase.coprocessor.AggregateImplementation");

		//AggregationClient封装了一系列hbase的聚合函数
		AggregationClient aggregationClient = new AggregationClient(conf);
		Scan scan = new Scan();
		//列解析器，用于解析列中的值，注意解析器类型需要和hbase存储值的类型一致，hbase提供Long、Double、BigDecimal三种类型的解析器。
		// ColumnInterpreter ci = new LongColumnInterpreter();
		ColumnInterpreter ci = new MyStringIntColumnInterpreter();
		TableName tn = TableName.valueOf("users");
		//rowCount可以不指定列族
		long count = aggregationClient.rowCount(tn, ci ,scan);
		//1936
		System.out.println(count);
		// scan.addFamily(Bytes.toBytes("int"));
		scan.addColumn(Bytes.toBytes("salaryDet"), Bytes.toBytes("gross"));
		//max必须指定列族，
		Object result = aggregationClient.max(tn, ci, scan);
		System.out.println(result);
		Object l = aggregationClient.sum(tn, ci, scan);
		System.out.println(l);
		double d = aggregationClient.avg(tn, ci, scan);
		System.out.println(d);
		//中位数
		Object l2 = aggregationClient.median(tn, ci, scan);
		System.out.println(l2);
		//标准差
		Object l3 = aggregationClient.std(tn, ci, scan);
		System.out.println(l3);
		aggregationClient.close();

	}
}
